package com.thiswhyme.mq.kafka.consumer2.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ContainerProperties;

/**
 * @description: Kafka配置
 * @author: thiswhyme
 * @date: 2021/4/21
 */
@Slf4j
@Configuration
public class KafkaConfig {

    @Value("${spring.application.name}")
    private String appName;

    /**
     * 自定义KafkaListenerContainerFactory，用于过滤消息
     * @param consumerFactory
     * @return
     */
    @Bean("filterContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> filterContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);

        // 消息过滤策略，返回true消息则被过滤
        factory.setRecordFilterStrategy(consumerRecord -> {
            log.info("====进入消息过滤器, {}", consumerRecord.toString());
            return !appName.equals(consumerRecord.key());
        });
        // 被过滤的消息是否丢弃，手动ack模式时才会生效
        factory.setAckDiscarded(true);
        return factory;
    }

    /**
     * 自定义KafkaListenerContainerFactory，用于手动确认
     * @param consumerFactory
     * @return
     */
    @Bean("manualAckContainerFactory")
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> manualAckContainerFactory(
            ConsumerFactory<String, String> consumerFactory) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL);
        return factory;
    }
}
